package com.seedog.mq.distributed;

import com.seedog.mq.model.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消息复制器实现类，提供分布式环境中的消息复制功能
 */
public class MessageReplicatorImpl implements MessageReplicator {
    private static final Logger logger = LoggerFactory.getLogger(MessageReplicatorImpl.class);
    
    private DistributedConfig config;
    private ClusterManager clusterManager;
    private MessageAckCallback messageAckCallback;
    private final ExecutorService replicationExecutor = Executors.newFixedThreadPool(10);
    private final AtomicBoolean running = new AtomicBoolean(false);
    private final Map<String, ReplicationStatus> messageReplicationStatus = new ConcurrentHashMap<>();
    private final ScheduledExecutorService cleanupScheduler = Executors.newScheduledThreadPool(1);
    
    @Override
    public void initialize(DistributedConfig config, ClusterManager clusterManager) {
        if (config == null) {
            throw new IllegalArgumentException("DistributedConfig cannot be null");
        }
        
        if (clusterManager == null) {
            throw new IllegalArgumentException("ClusterManager cannot be null");
        }
        
        this.config = config;
        this.clusterManager = clusterManager;
        
        // 注册消息处理器，处理来自其他节点的消息复制请求
        clusterManager.registerMessageHandler("REPLICATE_MESSAGE", this::handleReplicateMessage);
        clusterManager.registerMessageHandler("MESSAGE_ACK", this::handleMessageAck);
        clusterManager.registerMessageHandler("SYNC_MESSAGES", this::handleSyncMessages);
        
        logger.info("MessageReplicator initialized with replication factor: {}", config.getReplicationFactor());
    }
    
    @Override
    public boolean replicateMessage(Message message) {
        checkRunning();
        
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }
        
        // 如果当前节点不是主节点，不处理消息复制
        if (!clusterManager.isMaster()) {
            logger.warn("Only master node can initiate message replication");
            return false;
        }
        
        // 获取要复制到的目标节点
        List<Node> targetNodes = selectTargetNodes();
        if (targetNodes.isEmpty()) {
            logger.warn("No target nodes available for replication");
            return false;
        }
        
        // 创建消息复制状态
        ReplicationStatus status = new ReplicationStatus(message.getMessageId(), targetNodes.size());
        messageReplicationStatus.put(message.getMessageId(), status);
        
        logger.debug("Starting replication for message {} to {} nodes", 
                message.getMessageId(), targetNodes.size());
        
        // 并行复制到目标节点
        for (Node node : targetNodes) {
            replicationExecutor.submit(() -> replicateToNode(node, message));
        }
        
        // 立即返回成功，实际复制结果通过回调通知
        return true;
    }
    
    @Override
    public int replicateMessages(List<Message> messages) {
        checkRunning();
        
        if (messages == null || messages.isEmpty()) {
            throw new IllegalArgumentException("Messages cannot be null or empty");
        }
        
        int successfulReplications = 0;
        
        for (Message message : messages) {
            if (replicateMessage(message)) {
                successfulReplications++;
            }
        }
        
        return successfulReplications;
    }
    
    @Override
    public boolean syncMessages(String nodeId, List<Message> messages) {
        checkRunning();
        
        if (nodeId == null || nodeId.isEmpty()) {
            throw new IllegalArgumentException("Node ID cannot be null or empty");
        }
        
        if (messages == null || messages.isEmpty()) {
            return true; // 空消息列表被视为成功同步
        }
        
        logger.info("Syncing {} messages to node {}", messages.size(), nodeId);
        
        // 构建同步消息请求
        SyncMessagesRequest request = new SyncMessagesRequest(messages);
        
        // 发送同步请求到目标节点
        boolean success = clusterManager.sendMessageToNode(nodeId, request);
        
        if (success) {
            logger.debug("Successfully sent sync request to node {}", nodeId);
        } else {
            logger.warn("Failed to send sync request to node {}", nodeId);
        }
        
        return success;
    }
    
    @Override
    public List<Message> fetchMessagesFromNode(String nodeId) {
        checkRunning();
        
        if (nodeId == null || nodeId.isEmpty()) {
            throw new IllegalArgumentException("Node ID cannot be null or empty");
        }
        
        logger.info("Fetching messages from node {}", nodeId);
        
        // 简化实现：在实际分布式系统中，这里应该发送请求并等待响应
        // 模拟从节点获取消息
        return new ArrayList<>();
    }
    
    @Override
    public void setMessageAckCallback(MessageAckCallback callback) {
        this.messageAckCallback = callback;
    }
    
    @Override
    public void start() {
        if (running.compareAndSet(false, true)) {
            logger.info("Starting MessageReplicator...");
            
            // 启动清理过期复制状态的任务
            cleanupScheduler.scheduleAtFixedRate(this::cleanupExpiredStatus,
                    5, 5, TimeUnit.MINUTES);
            
            logger.info("MessageReplicator started successfully");
        }
    }
    
    @Override
    public void stop() {
        if (running.compareAndSet(true, false)) {
            logger.info("Stopping MessageReplicator...");
            
            replicationExecutor.shutdown();
            cleanupScheduler.shutdown();
            
            try {
                if (!replicationExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                    replicationExecutor.shutdownNow();
                }
                if (!cleanupScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    cleanupScheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                replicationExecutor.shutdownNow();
                cleanupScheduler.shutdownNow();
            }
            
            messageReplicationStatus.clear();
            logger.info("MessageReplicator stopped successfully");
        }
    }
    
    // 选择要复制的目标节点
    private List<Node> selectTargetNodes() {
        List<Node> activeNodes = clusterManager.getActiveNodes();
        List<Node> targetNodes = new ArrayList<>();
        
        // 过滤掉当前节点
        for (Node node : activeNodes) {
            if (!node.getNodeId().equals(clusterManager.getCurrentNode().getNodeId())) {
                targetNodes.add(node);
            }
        }
        
        // 如果节点数量超过复制因子，随机选择部分节点
        int replicationFactor = Math.min(config.getReplicationFactor(), targetNodes.size());
        if (replicationFactor < targetNodes.size()) {
            Collections.shuffle(targetNodes);
            targetNodes = targetNodes.subList(0, replicationFactor);
        }
        
        return targetNodes;
    }
    
    // 复制消息到指定节点
    private void replicateToNode(Node node, Message message) {
        try {
            logger.trace("Replicating message {} to node {}", message.getMessageId(), node.getNodeId());
            
            // 构建复制消息请求
            ReplicateMessageRequest request = new ReplicateMessageRequest(message);
            
            // 发送复制请求到目标节点
            boolean success = clusterManager.sendMessageToNode(node.getNodeId(), request);
            
            if (success) {
                logger.debug("Successfully sent message {} to node {}", message.getMessageId(), node.getNodeId());
            } else {
                logger.warn("Failed to send message {} to node {}", message.getMessageId(), node.getNodeId());
                
                // 处理复制失败
                handleReplicationFailure(message.getMessageId(), "Failed to send to node: " + node.getNodeId());
            }
        } catch (Exception e) {
            logger.error("Error replicating message {} to node {}", 
                    message.getMessageId(), node.getNodeId(), e);
            
            // 处理复制失败
            handleReplicationFailure(message.getMessageId(), "Exception: " + e.getMessage());
        }
    }
    
    // 处理复制消息请求
    private void handleReplicateMessage(String senderNodeId, Object message) {
        try {
            if (message instanceof ReplicateMessageRequest) {
                ReplicateMessageRequest request = (ReplicateMessageRequest) message;
                Message msg = request.getMessage();
                
                logger.debug("Received replicate message request from node {}: {}", 
                        senderNodeId, msg.getMessageId());
                
                // 存储消息（这里应该调用消息队列的存储方法）
                // 简化实现：实际系统中应该将消息存储到本地存储中
                
                // 发送确认消息
                sendMessageAck(senderNodeId, msg.getMessageId(), true);
            } else {
                logger.warn("Invalid replicate message request type from node {}", senderNodeId);
            }
        } catch (Exception e) {
            logger.error("Error handling replicate message request from node {}", senderNodeId, e);
            
            // 如果是ReplicateMessageRequest但处理失败，发送失败确认
            if (message instanceof ReplicateMessageRequest) {
                ReplicateMessageRequest request = (ReplicateMessageRequest) message;
                sendMessageAck(senderNodeId, request.getMessage().getMessageId(), false);
            }
        }
    }
    
    // 处理消息确认
    private void handleMessageAck(String senderNodeId, Object message) {
        try {
            if (message instanceof MessageAck) {
                MessageAck ack = (MessageAck) message;
                
                logger.debug("Received message ack from node {} for message {}", 
                        senderNodeId, ack.getMessageId());
                
                if (ack.isSuccess()) {
                    // 处理成功确认
                    handleAckSuccess(ack.getMessageId());
                } else {
                    // 处理失败确认
                    handleReplicationFailure(ack.getMessageId(), ack.getReason());
                }
            } else {
                logger.warn("Invalid message ack type from node {}", senderNodeId);
            }
        } catch (Exception e) {
            logger.error("Error handling message ack from node {}", senderNodeId, e);
        }
    }
    
    // 处理同步消息请求
    private void handleSyncMessages(String senderNodeId, Object message) {
        try {
            if (message instanceof SyncMessagesRequest) {
                SyncMessagesRequest request = (SyncMessagesRequest) message;
                List<Message> messages = request.getMessages();
                
                logger.debug("Received sync messages request from node {} with {} messages", 
                        senderNodeId, messages.size());
                
                // 存储同步的消息（这里应该调用消息队列的存储方法）
                // 简化实现：实际系统中应该将消息存储到本地存储中
                
                // 发送确认消息
                sendMessageAck(senderNodeId, "SYNC_COMPLETED", true);
            } else {
                logger.warn("Invalid sync messages request type from node {}", senderNodeId);
            }
        } catch (Exception e) {
            logger.error("Error handling sync messages request from node {}", senderNodeId, e);
        }
    }
    
    // 处理确认成功
    private void handleAckSuccess(String messageId) {
        ReplicationStatus status = messageReplicationStatus.get(messageId);
        if (status != null) {
            int remaining = status.decrementRemaining();
            logger.debug("Message {} replication remaining: {}", messageId, remaining);
            
            // 如果所有复制都已完成，触发回调
            if (remaining == 0) {
                messageReplicationStatus.remove(messageId);
                
                if (messageAckCallback != null) {
                    messageAckCallback.onMessageReplicated(messageId, status.getTotalNodes());
                }
            }
        }
    }
    
    // 处理复制失败
    private void handleReplicationFailure(String messageId, String reason) {
        // 简化实现：在实际系统中，这里应该有重试机制和故障处理逻辑
        
        ReplicationStatus status = messageReplicationStatus.get(messageId);
        if (status != null) {
            // 检查是否达到最小复制数量要求
            int successfulReplications = status.getTotalNodes() - status.getRemaining();
            int minRequiredReplications = Math.max(1, config.getReplicationFactor() / 2);
            
            if (successfulReplications >= minRequiredReplications) {
                // 即使有部分节点复制失败，但满足最小复制数量要求，仍视为成功
                messageReplicationStatus.remove(messageId);
                
                if (messageAckCallback != null) {
                    messageAckCallback.onMessageReplicated(messageId, successfulReplications);
                }
            } else {
                // 不满足最小复制数量要求，视为失败
                messageReplicationStatus.remove(messageId);
                
                if (messageAckCallback != null) {
                    messageAckCallback.onMessageReplicationFailed(messageId, reason);
                }
            }
        }
    }
    
    // 发送消息确认
    private void sendMessageAck(String targetNodeId, String messageId, boolean success) {
        MessageAck ack = new MessageAck(messageId, success);
        clusterManager.sendMessageToNode(targetNodeId, ack);
    }
    
    // 清理过期的复制状态
    private void cleanupExpiredStatus() {
        long currentTime = System.currentTimeMillis();
        long expirationTime = 5 * 60 * 1000; // 5分钟
        
        for (Iterator<Map.Entry<String, ReplicationStatus>> it = messageReplicationStatus.entrySet().iterator(); it.hasNext();) {
            Map.Entry<String, ReplicationStatus> entry = it.next();
            ReplicationStatus status = entry.getValue();
            
            if (currentTime - status.getCreationTime() > expirationTime) {
                logger.debug("Cleaning up expired replication status for message {}", entry.getKey());
                it.remove();
                
                // 通知回调
                if (messageAckCallback != null) {
                    messageAckCallback.onMessageReplicationFailed(
                            entry.getKey(), 
                            "Replication timeout after " + expirationTime + "ms"
                    );
                }
            }
        }
    }
    
    // 检查消息复制器是否正在运行
    private void checkRunning() {
        if (!running.get()) {
            throw new IllegalStateException("MessageReplicator is not running");
        }
    }
    
    // 复制消息请求内部类
    private static class ReplicateMessageRequest {
        private final Message message;
        
        public ReplicateMessageRequest(Message message) {
            this.message = message;
        }
        
        public Message getMessage() {
            return message;
        }
    }
    
    // 消息确认内部类
    private static class MessageAck {
        private final String messageId;
        private final boolean success;
        private final String reason;
        
        public MessageAck(String messageId, boolean success) {
            this(messageId, success, null);
        }
        
        public MessageAck(String messageId, boolean success, String reason) {
            this.messageId = messageId;
            this.success = success;
            this.reason = reason;
        }
        
        public String getMessageId() {
            return messageId;
        }
        
        public boolean isSuccess() {
            return success;
        }
        
        public String getReason() {
            return reason;
        }
    }
    
    // 同步消息请求内部类
    private static class SyncMessagesRequest {
        private final List<Message> messages;
        
        public SyncMessagesRequest(List<Message> messages) {
            this.messages = messages;
        }
        
        public List<Message> getMessages() {
            return messages;
        }
    }
    
    // 复制状态内部类
    private static class ReplicationStatus {
        private final String messageId;
        private final int totalNodes;
        private final AtomicInteger remaining;
        private final long creationTime;
        
        public ReplicationStatus(String messageId, int totalNodes) {
            this.messageId = messageId;
            this.totalNodes = totalNodes;
            this.remaining = new AtomicInteger(totalNodes);
            this.creationTime = System.currentTimeMillis();
        }
        
        public String getMessageId() {
            return messageId;
        }
        
        public int getTotalNodes() {
            return totalNodes;
        }
        
        public int getRemaining() {
            return remaining.get();
        }
        
        public int decrementRemaining() {
            return remaining.decrementAndGet();
        }
        
        public long getCreationTime() {
            return creationTime;
        }
    }
}